package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.ZoneId;
//卢顺祥  用户域用户注册统计事实表
public class DwdUserRegister {
    public static void main(String[] args) throws Exception {

        // TODO 1. 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.getConfig().setLocalTimeZone(ZoneId.of("GMT+8"));


        // TODO 3. 从 Kafka 读取 topic_db 数据，封装为 Flink SQL 表
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_user_register_group"));

        // TODO 4. 读取用户表数据
        Table userInfo = tableEnv.sqlQuery("" +
                "select\n" +
                "data['id'] user_id,\n" +
                "data['login_name'] login_name,\n" +
                "data['phone_num'] phone_num,\n" +
                "data['email'] email,\n" +
                "data['user_level'] user_level,\n" +
                "data['birthday'] birthday,\n" +
                "ts\n" +
                "from topic_db\n" +
                "where `table` = 'user_info'\n" +
                "and `type` = 'insert'");
        tableEnv.createTemporaryView("user_info", userInfo);

        // tableEnv.executeSql("select * from user_info ").print();

        // TODO 5. 创建 Upsert-Kafka dwd_user_register 表
        tableEnv.executeSql("" +
                "create table dwd_user_register(\n" +
                "user_id string,\n" +
                "login_name string,\n" +
                "phone_num string,\n" +
                "email string,\n" +
                "user_level string,\n" +
                "birthday string,\n" +
                "ts string,\n" +
                "primary key(user_id) not enforced\n" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_user_register"));

        // TODO 6. 将输入写入 Upsert-Kafka 表
        tableEnv.executeSql("insert into dwd_user_register select * from user_info");
    }
}
